//
//  stats_mcast_server.cpp
// ~~~~~~~~~~~~~~~~~~~~~
//
// Copyright (c) 2013-2014 Razvan Coca (razvan dot coca at  gmail dot com)
// 

#include "components/defs/handler_types.hpp"
#include "stats_mcast_server.hpp"

stats_mcast_server::stats_mcast_server(ba::io_service& ios, udp_endpoint_t& local_if, udp_endpoint_t& mgroup, int ttl,
                                       crypto_buf_t& iv, crypto_buf_t& key,
                                       bt::seconds persist_interval,
                                       bt::minutes expected_lifetime,
                                       const std::string &persist_path, int threads, const std::string &cluster, const std::string &node):
    ios_(ios),
    persist_interval_(persist_interval),
    processor_fsm_(ios,persist_path,expected_lifetime,cluster,node),
    threads_(threads),
    service_work_(new ba::io_service::work(ios))
{
    mcast_service_process_=components::service::mcast_service_process::create(ios,local_if,mgroup,ttl,iv,key,message_processor_);
    message_processor_.add(boost::bind(&components::protocol::stats_server_processor::process_request_get,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_server_processor::process_request_set,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_server_processor::process_do_reply,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_server_processor::process_end,boost::addressof(processor_fsm_),_1,_2,_3));
}

void thread_func(ba::io_service& ios)
{
    while(true)
    {
        try{
            std::size_t qlen=ios.run();
            LOG(ERROR)<<"service queue len:"<<qlen;
            if(!qlen)break;
        }
        catch(shutdown_exception const& e)
        {
            LOG(ERROR)<<"service thread error:"<<e.what();
            break;
        }
        catch(std::exception  const& e)
        {
            LOG(ERROR)<<"service thread error:"<<e.what();           
        }
    }
}

bool stats_mcast_server::run()
{
    for(int i=0;i<threads_;i++)
        thread_group_.add_thread(new boost::thread(boost::bind(&thread_func,boost::ref(ios_))));

    phdl::system_crypto_handler_t
            handler=boost::bind(&stats_mcast_server::handle_service_done,this,ba::placeholders::error,_2);

    return mcast_service_process_->exec(handler);
}

bool stats_mcast_server::cancel()
{
    return mcast_service_process_->cancel();
}

bool stats_mcast_server::shutdown()
{
    auto killer=[](){throw shutdown_exception("Shutdown requested");};
    service_work_.reset();
    bool ret=cancel();
    mcast_service_process_.reset();
    ret &= processor_fsm_.shutdown();
    for(int i=0;i<2*threads_;i++) ios_.post(killer);
    thread_group_.join_all();
    return ret;
}

void stats_mcast_server::handle_service_done (bs::error_code const& ec,const crypto_error& eec)
{
    if(!phdl::is_error(eec))
    {

    }
}
